#include "Avalon/IO/TaskWriter.hpp"
#include "Avalon/IO/Buffer.hpp"
#include "Avalon/Threading/Task.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::posix_time;
using namespace std;

class TaskWriter::WriterTask : public Task {
  public:

    WriterTask(TaskWriter* writer)
        : m_writer(writer) {}

    virtual ~WriterTask() {}

  protected:

    virtual void ProcessTask() {
      vector<pair<Buffer, AsyncOwner<void>::Ptr> > pendingWrites;
      {
        lock_guard<mutex> lock(m_writer->m_mutex);
        std::swap(pendingWrites, m_writer->m_pendingWrites);
      }
      for(vector<pair<Buffer, AsyncOwner<void>::Ptr> >::const_iterator i =
          pendingWrites.begin(); i != pendingWrites.end(); ++i) {
        Async<void>::Ptr writeResult = m_writer->m_destination->Write(i->first);
        writeResult->Wait(pos_infin);
        if(writeResult->GetState() == BaseAsync::EXCEPTION) {
          i->second->CopyException(writeResult);
        } else {
          i->second->SetComplete();
        }
      }
    }

  private:
    TaskWriter* m_writer;
};

TaskWriter::TaskWriter(Carry<Writer> destination)
    : m_destination(destination) {
  m_task.reset(new WriterTask(this));
}

TaskWriter::~TaskWriter() {
  m_task->Wait();
}

Async<void>::Ptr TaskWriter::Write(const void* data, int size) {
  Buffer copy;
  copy.Append(data, size);
  return Write(copy);
}

Async<void>::Ptr TaskWriter::Write(const Buffer& data) {
  AsyncOwner<void>::Ptr result = AsyncOwner<void>::Create();
  lock_guard<mutex> lock(m_mutex);
  m_pendingWrites.push_back(make_pair(data, result));
  if(m_pendingWrites.size() == 1) {
    m_task->Enqueue();
  }
  return result;
}
